package com.lagou.edu.consumer;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.lagou.edu.decoder.RpcDecoder;
import com.lagou.edu.encoder.RpcEncoder;
import com.lagou.edu.handler.ClientHandler;
import com.lagou.edu.pojo.ServiceNodeInfo;
import com.lagou.edu.request.RpcRequest;
import com.lagou.edu.response.RpcResponse;
import com.lagou.edu.serializer.JSONSerializer;
import com.lagou.edu.zookeeper.ZookeeperFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;

import java.lang.reflect.Proxy;
import java.nio.charset.StandardCharsets;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author 赵虎
 * @date 2020/11/25 12:06
 */
public class RpcConsumer {
    private static final ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static final Map<String, Map<String, ClientHandler>> serviceMap = Maps.newHashMap();

    private static final ZookeeperFactory zookeeperFactory = ZookeeperFactory.getInstance();

    private void initClient(String className) throws Exception {

        List<ServiceNodeInfo> serviceNodeInfoList = zookeeperFactory.getChildNodeValue(className);

        if (serviceNodeInfoList == null || serviceNodeInfoList.size() == 0) {
            System.out.println("服务端未启动");
            return;
        }

        Map<String, ClientHandler> clientHandlers = Maps.newHashMap();
        for (ServiceNodeInfo serviceNodeInfo : serviceNodeInfoList) {
            ClientHandler clientHandler = getClientHandler(className, serviceNodeInfo.getIp(), serviceNodeInfo.getPort());
            clientHandlers.put(clientHandler.getServicePath(), clientHandler);
        }

        serviceMap.put(className, clientHandlers);

        PathChildrenCache pathChildrenCache = zookeeperFactory.getPathChildrenCache(className);
        pathChildrenCache.getListenable().addListener((client, event) -> {

            ServiceNodeInfo serviceNodeInfo = JSON.parseObject(new String(event.getData().getData(), StandardCharsets.UTF_8), ServiceNodeInfo.class);

            Map<String, ClientHandler> clientHandlerMap = serviceMap.get(className);

            ClientHandler clientHandler = null;
            switch (event.getType()) {
                case CHILD_ADDED:

                    clientHandler = getClientHandler(className, serviceNodeInfo.getIp(), serviceNodeInfo.getPort());

                    clientHandlerMap.put(serviceNodeInfo.getPath(), clientHandler);

                    serviceMap.put(className, clientHandlerMap);

                    System.out.println("添加子节点:" + serviceNodeInfo);
                    break;
                case CHILD_REMOVED:
                    clientHandler = clientHandlerMap.get(serviceNodeInfo.getPath());

                    clientHandler.close();

                    clientHandlerMap.remove(serviceNodeInfo.getPath());

                    serviceMap.put(className, clientHandlerMap);

                    System.out.println("删除子节点:" + serviceNodeInfo);
                default:
                    break;
            }
        });
    }

    public ClientHandler choseClientHandler(String className) throws Exception {
        ServiceNodeInfo serviceNodeInfo = zookeeperFactory.getChildNodeValue(className)
                .stream().min(Comparator.comparing(ServiceNodeInfo::getUseTime))
                .orElse(null);
        if (serviceNodeInfo != null) {
            Map<String, ClientHandler> clientHandlerMap = serviceMap.get(className);
            return clientHandlerMap.get(serviceNodeInfo.getPath());
        }
        return null;
    }

    private static ClientHandler getClientHandler(String className, String ip, int port) throws InterruptedException {
        ClientHandler clientHandler = new ClientHandler();

        NioEventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(clientHandler);
                    }
                });
        bootstrap.connect(ip, port).sync();

        String servicePath = StringUtils.join(ZookeeperFactory.urlSubMark, className, ZookeeperFactory.urlSubMark, ip, ZookeeperFactory.addressSubMark, port);

        //设置客户端连接服务的zk路径
        clientHandler.setServicePath(servicePath);

        //设置zk工厂
        clientHandler.setZookeeperFactory(zookeeperFactory);
        return clientHandler;
    }


    public Object createProxy(final Class<?> clazz) {

        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{clazz}, (proxy, method, args) -> {
            RpcRequest request = new RpcRequest();
            request.setClassName(clazz.getSimpleName());
            request.setMethodName(method.getName());
            request.setParameters(args);
            request.setParameterTypes(method.getParameterTypes());
            request.setResultTypes(method.getReturnType());

            if (serviceMap.size() == 0) {
                initClient(clazz.getName());
            }

            ClientHandler clientHandler = choseClientHandler(clazz.getName());

            if (clientHandler != null) {
                clientHandler.setRpcRequest(request);
                return executor.submit(clientHandler).get();
            } else {
                throw new Exception("没有可用服务端");
            }
        });
    }
}
